#include "MsgServer.h"
#include <boost/assert.hpp>

namespace RobotControl{
namespace Common{

/// \brief 组件构造
MsgServer::MsgServer(zmq::context_t *ctx)
{
	// 如果m_cmx不正确， addUsb和addRep会抛出异常的.
	m_ctx = ctx;
}

/// \brief 添加订阅
/// \param[in] topic: 订阅消息
/// \param[in] callback: 回调方法
bool MsgServer::addSubscriber(const std::string &topic,const SubCallback_t &callback)
{
	zmq::socket_t socket (*m_ctx, ZMQ_SUB);
	socket.connect (topic);
	socket.setsockopt(ZMQ_SUBSCRIBE, "", 0);
	m_sockets.emplace_back(std::move(socket));
    m_types.emplace_back(Sub);
    m_callbacks.emplace_back(std::make_tuple(callback, RepCallback_t()));
	return true;
}

/// \brief 添加应答
/// \param[in] topic: 响应消息
/// \param[in] callback: 回调方法
bool MsgServer::addReply(const std::string &topic,const RepCallback_t &callback)
{
	zmq::socket_t socket (*m_ctx, ZMQ_REP);
	socket.bind (topic);
	m_sockets.emplace_back(std::move(socket));
    m_types.emplace_back(Rep);
    m_callbacks.emplace_back(std::make_tuple(SubCallback_t(), callback));
	return true;
}

/// \brief 启动
bool MsgServer::start()
{
	m_pollers = new zmq_pollitem_t[m_sockets.size()];
	for (int i=0; i<m_sockets.size(); i++){
		m_pollers[i] = {(void *)m_sockets[i], 0, ZMQ_POLLIN, 0};
	}
	return true;
}

/// \brief 运行消息回调, 处理消息的订阅和回调
void MsgServer::run(const int32_t timeout)
{
	if (m_sockets.size() <= 0) return;

	int ret = zmq::poll(m_pollers, m_sockets.size(), timeout);

	for (int i=0; i<m_sockets.size(); i++){
		if(m_pollers[i].revents & ZMQ_POLLIN) {
			if (m_types[i] == Sub){
				// 回调订阅
				zmq::message_t message;
				m_sockets[i].recv(&message);
				std::get<0>(m_callbacks[i])(message);
			}
			else if (m_types[i] == Rep){
				// 回调请求
				zmq::message_t req, rep;
				m_sockets[i].recv(&req);
				std::get<1>(m_callbacks[i])(req, rep);
				m_sockets[i].send(rep);
			}
		}
	}

}

}	// Common
}	// RobotControl
